package com.demo.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

@Slf4j
public class KafkaConsumerDemo {
    private final static List<String> contentKey = Arrays.asList(
            "imei", "time", "lng", "lat", "speed",
            "distance", "course", "voltage", "pyblank", "totalMileage",
            "gpsStatus", "createTime", "lbsList", "locTyp", "locProperties",
            "altitude", "precision");

    private static Consumer getConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.16.18.122:9092,172.16.18.123:9092,172.16.18.124:9092,172.16.18.125:9092");
//        props.put("group.id", "local-gps-trace");
        props.put("group.id", "df_can_detail_test");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        return consumer;
    }

    public static void main(String[] args) {
        Consumer consumer = getConsumer();
        consumer.subscribe(Collections.singletonList("topic_canDetail"));
//        consumer.subscribe(Collections.singletonList("topic_gps_trace"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {

                    JSONObject msgJson = JSON.parseObject(record.value());

                    Integer type = msgJson.getInteger("type");
//                    String niuju = msgJson.getString("x702C");
                    String niuju = msgJson.getString("x7307");
                    // x7302 被限制的扭矩  0x7307 smart扭矩限制
//                    StringBuilder sb = new StringBuilder(512);
//
//                    for (String key : contentKey) {
//                        sb.append(msgJson.getString(key)).append("\t");
//                    }
//                    String toString = sb.toString();
//                    String substring = toString.substring(0, toString.length() - 1);
//
//                    System.out.println(substring);

                    if (niuju != null) {
                        System.out.println(msgJson.toJSONString());
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }
}
